TwitterStreamAgent prevent connection in use and restart issues
Ensures the database connection is checked back in to the pool before restarting.
Restart TwitterStreamAgent outside of the Rufus::Scheduler work thread.
Dominik Sander
2 changed files with 9 additions and 6 deletions
+ 3
- 1
app/concerns/long_runnable.rb
@@ -98,13 +98,15 @@ module LongRunnable |
||
| 98 | 98 |
|
| 99 | 99 |
def terminate_thread! |
| 100 | 100 |
if thread |
| 101 |
- thread.terminate |
|
| 101 |
+ thread.instance_eval { ActiveRecord::Base.connection_pool.release_connection }
|
|
| 102 | 102 |
thread.wakeup if thread.status == 'sleep' |
| 103 |
+ thread.terminate |
|
| 103 | 104 |
end |
| 104 | 105 |
end |
| 105 | 106 |
|
| 106 | 107 |
def restart! |
| 107 | 108 |
without_alive_check do |
| 109 |
+ puts "--> Restarting #{id} at #{Time.now} <--"
|
|
| 108 | 110 |
stop! |
| 109 | 111 |
setup!(scheduler, mutex) |
| 110 | 112 |
run! |
+ 6
- 5
app/models/agents/twitter_stream_agent.rb
@@ -157,16 +157,14 @@ module Agents |
||
| 157 | 157 |
def setup |
| 158 | 158 |
require 'twitter/json_stream' |
| 159 | 159 |
@filter_to_agent_map = @config[:filter_to_agent_map] |
| 160 |
- |
|
| 161 |
- schedule_in RELOAD_TIMEOUT do |
|
| 162 |
- puts "--> Restarting TwitterStream #{id} at #{Time.now} <--"
|
|
| 163 |
- restart! |
|
| 164 |
- end |
|
| 165 | 160 |
end |
| 166 | 161 |
|
| 167 | 162 |
def run |
| 168 | 163 |
@recent_tweets = [] |
| 169 | 164 |
EventMachine.run do |
| 165 |
+ EventMachine.add_periodic_timer(RELOAD_TIMEOUT) do |
|
| 166 |
+ restart! |
|
| 167 |
+ end |
|
| 170 | 168 |
stream!(@filter_to_agent_map.keys, @agent) do |status| |
| 171 | 169 |
handle_status(status) |
| 172 | 170 |
end |
@@ -200,6 +198,9 @@ module Agents |
||
| 200 | 198 |
|
| 201 | 199 |
stream.on_error do |message| |
| 202 | 200 |
STDERR.puts " --> Twitter error: #{message} at #{Time.now} <--"
|
| 201 |
+ STDERR.puts " --> Sleeping for 15 seconds" |
|
| 202 |
+ sleep 15 |
|
| 203 |
+ restart! |
|
| 203 | 204 |
end |
| 204 | 205 |
|
| 205 | 206 |
stream.on_no_data do |message| |